package com.surfilter.massdata.spark.output;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.output.impl.HiveOutput;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.JdbcUtil;

public class OracleOutputforSmcs extends HiveOutput {
	private static final Log log = LogFactory.getLog(OracleOutputforSmcs.class);
	private static final long serialVersionUID = 1L;
	private String table;
	private String tableDim;
	private String columns;
	private String values;
	
	@Override
	public void write(DataFrame result, Map<String, String> commandMap) {
		
		System.out.println("===========OracleOutputforSmcs "+table+" begin============");
		long start=System.currentTimeMillis();
		List<Row> list = result.collectAsList();
		//result.show();
		//天表插入
		insertTable(list);
		//暂时屏蔽后续重新写
		/*//周、月、季度、年表插入
		insertTableDim(list);*/
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("OracleOutputforSmcs "+table+" :exectime: "+min+" min............");
	}

	private void insertTableDim(List<Row> list) {
		if(StringUtils.isNotBlank(tableDim) && list.size() > 0){
			String[] tableDimSplit = tableDim.split(",");
			for(String table : tableDimSplit){
				Map<String,Long> mapSeq =  CommonUtils.getSeq(table, list.size());
				long id = mapSeq.get("minSeq");
				String sqlDim = "insert into "+table+ "("+columns+") "+values;
				String[] columnsSplit = columns.split(",");
				Connection conn = null;
				PreparedStatement ps = null;
				conn = JdbcUtil.getConnection();
				try {
					ps = conn.prepareStatement(sqlDim);
					conn.setAutoCommit(false);
					int n = 0;
					for (Row row : list) {
						ps.setObject(1,id++);//id
						for (int i = 1; i < row.size(); i++) {
							ps.setObject(i + 1,row.getAs(columnsSplit[i]));
						}
						ps.addBatch();
					
						
						if (++n % 5000 == 0) {
							ps.executeBatch();
							conn.commit();
						}
					}
					ps.executeBatch();
					conn.commit();
				} catch (SQLException e) {
					log.error("========="+table+"========");
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}finally{
					JdbcUtil.close(ps,conn);
				}
			}
		}
	}

	private void insertTable(List<Row> list) {
		
		CommonUtils.createSeqIfNotExists(table);
		
		String[] columnsSplit = columns.split(",");
		Connection conn = null;
		PreparedStatement ps = null;
		String sql = "insert into "+table+ "("+columns+") "+values;
		conn = JdbcUtil.getConnection();
		try {
			ps = conn.prepareStatement(sql);
			conn.setAutoCommit(false);
			int n = 0;
			for (Row row : list) {
				/*for (int i = 0; i < row.size(); i++) {
					ps.setObject(i + 1, row.getAs(columnsSplit[i]));
				}*/
				for (int i = 1; i < row.size(); i++) {
					ps.setObject(i, row.getAs(columnsSplit[i]));
				}
				ps.addBatch();

				if (++n % 5000 == 0) {
					ps.executeBatch();
					conn.commit();
				}
			}
			ps.executeBatch();
			conn.commit();
		} catch (Exception e) {
			log.error("========="+table+"========");
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps,conn);
		}
	}
}
